
// <copyright file="ConcurrentStoreTests.cs" company="Subhadra Software Solutions">
//   
// </copyright>
// <summary>
//   The concurrent store tests.
// </summary>

namespace SubhadraSolutions.Sharp.Utils.Tests.Collections.Concurrent
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;

    using Microsoft.VisualStudio.TestTools.UnitTesting;

    using SubhadraSolutions.Sharp.Utils.Collections.Concurrent;
    using SubhadraSolutions.Sharp.Utils.Collections.Generic;
    using SubhadraSolutions.Sharp.Utils.Diagnostics;
    using SubhadraSolutions.Sharp.Utils.Tests.Collections.Concurrent.Internal;

    /// <summary>
    /// The concurrent store tests.
    /// </summary>
    [TestClass]
    public class ConcurrentStoreTests
    {
        #region Fields

        /// <summary>
        /// The random.
        /// </summary>
        private readonly Random _random = new Random();

        #endregion Fields

        #region Public Methods and Operators

        /// <summary>
        /// The blocking mode test.
        /// </summary>
        [TestMethod]
        public void BlockingModeTest()
        {
            const int numberOfThreads = 100;
            const int itemsPerThread = 1000;
            const int delayInFlushing = 1000;
            const int delayBetweenWrites = 10;

            int bufferSize = 500 + this._random.Next(500);
            const int periodicFlushInterval = 2000;
            const int maxWaitTime = -1;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
            Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
        }

        /// <summary>
        /// The buffer_ full flush_ test.
        /// </summary>
        [TestMethod]
        public void Buffer_FullFlush_Test()
        {
            const int numberOfThreads = 10;
            const int itemsPerThread = 1000;

            const int bufferSize = 1000;
            const int periodicFlushInterval = -1;
            const int maxWaitTime = -1;
            const int delayInFlushing = 0;
            const int delayBetweenWrites = 0;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
            Assert.IsTrue(mStore.NumberOfFlushes == 10);
            Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
        }

        /// <summary>
        /// The count based_ random blocking_ test.
        /// </summary>
        [TestMethod]
        public void CountBased_RandomBlocking_Test()
        {
            int count = this._random.Next(20) + 5;
            for (int i = 0; i < count; i++)
            {
                this.RandomBlocking_Test();
            }
        }

        /// <summary>
        /// The force flush_ or_ dispose_ test.
        /// </summary>
        [TestMethod]
        public void ForceFlush_Or_Dispose_Test()
        {
            const int numberOfThreads = 5;
            const int itemsPerThread = 1000;

            const int bufferSize = 5001;
            const int periodicFlushInterval = -1;
            const int maxWaitTime = -1;
            const int delayInFlushing = 0;
            const int delayBetweenWrites = 0;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = null;
            var t = new Thread(delegate() { mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing); });
            t.Start();
            if (!t.Join(2000))
            {
                GeneralHelper.AbortThread(t);
                Assert.Fail("Force flush did not happen in a timely manner");
            }
            else
            {
                const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
                Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
                Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
                Assert.IsTrue(mStore.NumberOfFlushes == 1);
                Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
            }
        }

        /// <summary>
        /// The non blocking mode_ no skip_ test.
        /// </summary>
        [TestMethod]
        public void NonBlockingMode_NoSkip_Test()
        {
            const int numberOfThreads = 100;
            const int itemsPerThread = 1000;
            const int delayInFlushing = 1000;
            const int delayBetweenWrites = 10;

            int bufferSize = 500 + this._random.Next(500);
            const int periodicFlushInterval = 2000;
            const int maxWaitTime = 1100;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
            Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
        }

        /// <summary>
        /// The non blocking mode_ skip_ test.
        /// </summary>
        [TestMethod]
        public void NonBlockingMode_Skip_Test()
        {
            const int numberOfThreads = 100;
            const int itemsPerThread = 1000;
            const int delayInFlushing = 1000;
            const int delayBetweenWrites = 10;

            int bufferSize = 500 + this._random.Next(500);
            const int periodicFlushInterval = 2000;
            const int maxWaitTime = 500;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped > 0);
        }

        /// <summary>
        /// The periodic flush_ test.
        /// </summary>
        [TestMethod]
        public void PeriodicFlush_Test()
        {
            const int numberOfThreads = 2;
            const int itemsPerThread = 2;

            const int bufferSize = 1000;
            const int periodicFlushInterval = 2000;
            const int maxWaitTime = -1;
            const int delayInFlushing = 0;
            const int delayBetweenWrites = 3000;

            var counts = new int[numberOfThreads];
            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            const long itemsWrittenShouldBe = (long)numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
            Assert.IsTrue(mStore.NumberOfFlushes == 2);
            Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
        }

        /// <summary>
        /// The random blocking_ test.
        /// </summary>
        [TestMethod]
        public void RandomBlocking_Test()
        {
            int numberOfThreads = this._random.Next(200) + 1;
            int itemsPerThread = this._random.Next(10000) + 1;
            int bufferSize = this._random.Next(10000) + 1;
            int delayInFlushing = this._random.Next(100);
            const int maxWaitTime = -1;
            int periodicFlushInterval = this._random.Next(2000);
            int delayBetweenWrites = this._random.Next(100) - this._random.Next(100);
            if (delayBetweenWrites < 0)
            {
                delayBetweenWrites = 0;
            }

            var counts = new int[numberOfThreads];

            MetricsEnabledConcurrentStore<Pair<int, int>> mStore = this.performTest(counts, itemsPerThread, bufferSize, periodicFlushInterval, maxWaitTime, delayBetweenWrites, delayInFlushing);
            long itemsWrittenShouldBe = numberOfThreads * (long)itemsPerThread;
            Assert.IsTrue(mStore.NumberOfItemsAttemptedToWrite == itemsWrittenShouldBe && mStore.NumberOfItemsFlushed + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe && mStore.NumberOfItemsWritten + mStore.NumberOfItemsSkipped == itemsWrittenShouldBe);
            Assert.IsTrue(mStore.NumberOfItemsSkipped == 0);
            Assert.IsTrue(this.validateCounts(counts, itemsPerThread));
        }

        /// <summary>
        /// The time based_ random blocking_ test.
        /// </summary>
        [TestMethod]
        public void TimeBased_RandomBlocking_Test()
        {
            int durationInSeconds = this._random.Next(20) + 10;
            long durationInTicks = durationInSeconds * TimeSpan.TicksPerSecond;
            long ticksBefore = SharedStopWatch.ElapsedTicks;
            while (SharedStopWatch.ElapsedTicks - ticksBefore < durationInTicks)
            {
                this.RandomBlocking_Test();
            }
        }

        #endregion Public Methods and Operators

        #region Methods

        /// <summary>
        /// The perform test.
        /// </summary>
        /// <param name="counts">
        /// The counts.
        /// </param>
        /// <param name="itemsPerThread">
        /// The items per thread.
        /// </param>
        /// <param name="bufferSize">
        /// The buffer size.
        /// </param>
        /// <param name="periodicFlushInterval">
        /// The periodic flush interval.
        /// </param>
        /// <param name="maxWaitTime">
        /// The max wait time.
        /// </param>
        /// <param name="delayBetweenWrites">
        /// The delay between writes.
        /// </param>
        /// <param name="delayInFlushing">
        /// The delay in flushing.
        /// </param>
        /// <returns>
        /// The <see cref="MetricsEnabledConcurrentStore"/>.
        /// </returns>
        private MetricsEnabledConcurrentStore<Pair<int, int>> performTest(IList<int> counts, int itemsPerThread, int bufferSize, int periodicFlushInterval, int maxWaitTime, int delayBetweenWrites, int delayInFlushing)
        {
            var store = new ConcurrentStore<Pair<int, int>>(bufferSize, ThreadPriority.Normal, periodicFlushInterval, maxWaitTime);
            var mStore = new MetricsEnabledConcurrentStore<Pair<int, int>>(store);
            mStore.OnStoreFlush += delegate(object sender, StoreFlushEventArgs<Pair<int, int>> e)
                {
                    foreach (var item in e)
                    {
                        counts[item.First]++;

                        // System.Diagnostics.Debug.WriteLine(item.First.ExportAsString() + @"\t-\t" + item.Second.ExportAsString());
                    }

                    if (delayInFlushing > 0)
                    {
                        Thread.Sleep(delayInFlushing);
                    }
                };

            this.writeItems(mStore, counts, itemsPerThread, delayBetweenWrites);

            return mStore;
        }

        /// <summary>
        /// The thread start.
        /// </summary>
        /// <param name="state">
        /// The state.
        /// </param>
        private void threadStart(object state)
        {
            var info = (ConcurrentStoreTestInfo)state;
            info.handle.WaitOne();
            for (int i = info.from; i < info.to; i++)
            {
                info.store.Add(Pair<int, int>.CreateNew(info.threadIndex, i));
                if (info.delay > 0)
                {
                    Thread.Sleep(info.delay);
                }
            }
        }

        /// <summary>
        /// The validate counts.
        /// </summary>
        /// <param name="counts">
        /// The counts.
        /// </param>
        /// <param name="itemsPerThread">
        /// The items per thread.
        /// </param>
        /// <returns>
        /// The <see cref="bool"/>.
        /// </returns>
        private bool validateCounts(IEnumerable<int> counts, int itemsPerThread)
        {
            return counts.All(t => t == itemsPerThread);
        }

        /// <summary>
        /// The write items.
        /// </summary>
        /// <param name="store">
        /// The store.
        /// </param>
        /// <param name="counts">
        /// The counts.
        /// </param>
        /// <param name="itemsPerThread">
        /// The items per thread.
        /// </param>
        /// <param name="delayBetweenWrites">
        /// The delay between writes.
        /// </param>
        private void writeItems(IConcurrentStore<Pair<int, int>> store, ICollection<int> counts, int itemsPerThread, int delayBetweenWrites)
        {
            var mEvent = new ManualResetEvent(false);
            var threads = new Thread[counts.Count];
            for (int i = 0; i < threads.Length; i++)
            {
                threads[i] = new Thread(this.threadStart);
                var info = new ConcurrentStoreTestInfo { handle = mEvent, threadIndex = i, @from = i * itemsPerThread, to = (i + 1) * itemsPerThread, delay = delayBetweenWrites, store = store };
                threads[i].Start(info);
            }

            mEvent.Set();
            foreach (Thread thread in threads)
            {
                thread.Join();
            }

            store.Dispose();
        }

        #endregion Methods
    }
}
